package com.qssoft.cqry.synchdb.pool;

import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.ObjectUtil;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.QueryEventData;
import com.github.shyiko.mysql.binlog.event.RotateEventData;
import com.qssoft.cqry.synchdb.adapter.MysqlBinlogAdapter;
import com.qssoft.cqry.synchdb.adapter.MysqlBinlogPositionAdapter;
import com.qssoft.cqry.synchdb.properties.SynchDbProperties;
import com.qssoft.cqry.synchdb.vo.BinlogPositionVo;
import com.qssoft.cqry.synchdb.vo.MysqlBinlogVo;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.Arrays;

/**
 * @author liju.z
 */
@Slf4j
public class MysqlBinlogClient implements Runnable, SmartInitializingSingleton {
    @Autowired
    private MysqlBinlogAdapter redisMysqlBinlogAdapter;

    private MysqlBinlogPositionAdapter mysqlBinlogPositionAdapter;

    private SynchDbProperties.SynchMysqlProperties mysqlProperties;

    public MysqlBinlogClient(SynchDbProperties.SynchMysqlProperties mysqlProperties, MysqlBinlogPositionAdapter mysqlBinlogPositionAdapter) {
        this.mysqlProperties = mysqlProperties;
        this.mysqlBinlogPositionAdapter = mysqlBinlogPositionAdapter;
    }

    /**
     * 创建
     *
     * @return
     */
    public BinaryLogClient createMysqlBinlogClient() {
        return new BinaryLogClient(mysqlProperties.getIp(), mysqlProperties.getPort(), mysqlProperties.getUsername(), mysqlProperties.getPassword());
    }

    /**
     * 初始化设置mysql日志坐标
     */
    private void initBinlogPosition(BinaryLogClient binaryLogClient) {
        BinlogPositionVo binlogPositionVo = mysqlBinlogPositionAdapter.getBinlogPositionVo();
        if (binlogPositionVo == null) return;
        //mysql日志文件名称
        if (ObjectUtil.isNotEmpty(binlogPositionVo.getBinlogName())) {
            binaryLogClient.setBinlogFilename(binlogPositionVo.getBinlogName());
        }
        //mysql日志文件坐标
        if (ObjectUtil.isNotEmpty(binlogPositionVo.getPosition())) {
            binaryLogClient.setBinlogPosition(binlogPositionVo.getPosition());
        }
    }

    /**
     * 添加监听
     */
    private void addListener(BinaryLogClient binaryLogClient) {
        binaryLogClient.registerEventListener(event -> {
            EventType eventType = event.getHeader().getEventType();
            //记录开始监听日志文件状态
            if (EventType.ROTATE.equals(eventType)) {
                RotateEventData rotateEventData = event.getData();
                BinlogPositionVo binlogPositionVo = new BinlogPositionVo();
                binlogPositionVo.setBinlogName(rotateEventData.getBinlogFilename());
                binlogPositionVo.setPosition(rotateEventData.getBinlogPosition());
                binlogPositionVo.setServerId(event.getHeader().getServerId());
                if (log.isDebugEnabled()) {
                    log.debug("{}, {}, {}", rotateEventData.getBinlogFilename(), rotateEventData.getBinlogPosition(), event.getHeader().getServerId());
                }
                mysqlBinlogPositionAdapter.binlogPositionHandler(binlogPositionVo);
            }
            //MIXED模式
            else if (EventType.QUERY.equals(eventType)) {
                QueryEventData data = event.getData();
                if (ObjectUtil.equal(data.getSql(), "BEGIN")) {
                    if (log.isDebugEnabled()) {
                        log.debug("binlog BEGIN执行不记录...{}", data.toString());
                    }
                    return;
                }
                else if(ObjectUtil.equal(data.getSql(), "COMMIT")){
                    if (log.isDebugEnabled()) {
                        log.debug("binlog COMMIT执行不记录...{}", data.toString());
                    }
                    return;
                }
                else if (ObjectUtil.isNotEmpty(data.getSql())) {
                    //确认同步数据库表
                    if(ArrayUtil.isNotEmpty(mysqlProperties.getTables())){
                        long count = Arrays.stream(mysqlProperties.getTables()).filter(f -> data.getSql().indexOf(f) > 0).count();
                        if(count <= 0) {
                            if (log.isDebugEnabled()) {
                                log.debug("binlog 不在同步范围内...{}", data.getDatabase());
                            }
                            return;
                        }
                    }
                    try {
                        MysqlBinlogVo mysqlBinlogVo = new MysqlBinlogVo();
                        mysqlBinlogVo.setApplicationName(mysqlProperties.getApplicationName());
                        mysqlBinlogVo.setDatabase(data.getDatabase());
                        mysqlBinlogVo.setServerId(event.getHeader().getServerId());
                        mysqlBinlogVo.setQueryEventData(data);
                        redisMysqlBinlogAdapter.sqlHandler(mysqlBinlogVo);
                        if (log.isDebugEnabled()) {
                            log.debug("{},{},{}", event.getHeader().getServerId(), data.getDatabase(), data.getSql());
                        }
                        log.info("{},{},{}", event.getHeader().getServerId(), data.getDatabase(), data.getSql());
                    } catch (Exception ex) {
                        log.warn("执行sql语句出错...{}", ex);
                    }
                }
            }
        });
    }

    @SneakyThrows
    @Override
    public void run() {
        log.info("binlog >>> {}", mysqlProperties.getIp());
        BinaryLogClient mysqlBinlogClient = createMysqlBinlogClient();
        try {
            //初始化文件流坐标
            initBinlogPosition(mysqlBinlogClient);
            //添加监听事件
            addListener(mysqlBinlogClient);
            //创建socket链接
            mysqlBinlogClient.connect();
        } catch (IOException e) {
            log.error("同步数据库链接失败...{}", e);
            throw e;
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        new Thread(this).start();
    }
}
